package cn.gwm.flink.streaming.function.process;

import cn.gwm.flink.streaming.beans.estimateSoh.ConfigData;
import cn.gwm.flink.streaming.beans.estimateSoh.EstimateSohRecord;
import lombok.Data;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

/**
 * soh估算配置项广播流
 * @author GW00283474
 */
@Data
public class EstmateSohBroadcastProcessFunction extends KeyedBroadcastProcessFunction<String, EstimateSohRecord, Map<String, ConfigData>, EstimateSohRecord> {

    private MapStateDescriptor<String,ConfigData> configDataDesc;
    @Override
    public void processElement(EstimateSohRecord value, KeyedBroadcastProcessFunction<String, EstimateSohRecord, Map<String, ConfigData>, EstimateSohRecord>.ReadOnlyContext ctx, Collector<EstimateSohRecord> out) throws Exception {
       try {
           ReadOnlyBroadcastState<String, ConfigData> broadcastState = ctx.getBroadcastState(configDataDesc);
           Iterator<Map.Entry<String, ConfigData>> iterator = broadcastState.immutableEntries().iterator();
           HashMap<String,ConfigData> hashMap = new HashMap<>();
           while (iterator.hasNext()){
               Map.Entry<String, ConfigData> entry = iterator.next();
               if("1".equals(entry.getValue().getState())){
                   hashMap.put(entry.getKey(),entry.getValue());
               }
           }
           value.setConfigDataMap(hashMap);
           out.collect(value);
       }catch (Exception e){
           System.out.println("---------------EstmateSohBroadcastProcessFunction中出现异常");
           e.printStackTrace();
       }

    }

    @Override
    public void processBroadcastElement(Map<String, ConfigData> value, KeyedBroadcastProcessFunction<String, EstimateSohRecord, Map<String, ConfigData>, EstimateSohRecord>.Context ctx, Collector<EstimateSohRecord> out) throws Exception {
        BroadcastState<String, ConfigData> state = ctx.getBroadcastState(configDataDesc);
        state.clear();
        state.putAll(value);
    }

    public EstmateSohBroadcastProcessFunction(MapStateDescriptor<String, ConfigData> configDataDesc) {
        this.configDataDesc = configDataDesc;
    }

    public EstmateSohBroadcastProcessFunction(){
        super();
    }
}
